package org
import org.apache.spark.sql.SparkSession

object scala {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().master("local[*]").appName("spark").getOrCreate()
    val sc = spark.sparkContext
    //创建RDD 不指定分区就会打散
    val rdd1 = sc.makeRDD(List(1, 5, 3, 7, 6), 1)
    val rdd2 = sc.parallelize(List(('a', 90), ('b', 78), ('c', 66), ('b', 78)))
    //spark数据分析法
    val rdd3 = sc.makeRDD(List(1, 2, 3, 4, 5), 1)
    //交集intersection
    rdd1.intersection(rdd3).foreach(println)
    //并集union
    rdd1.union(rdd3).foreach(println)
    //差集subtract 站在rdd1的角度上看是 6 7
    rdd1.subtract(rdd3).foreach(println)
    //ZIP拉链 分区要一致 形成元组（key,value)
    rdd1.zip(rdd3).foreach(println)
    //笛卡尔积 1跟3中的每个原始都会重新组合成一组， 5*5=25组
    rdd1.cartesian(rdd3).foreach(println)
    //fiter rdd2中筛选出大于等于90分的数据
    //rdd2.filter(_._2>=90).foreach(println)
    //去重distinct
    // rdd2.distinct().foreach(println)
    //输出结果
    //  rdd1.foreach(println)
    // rdd2.collect().foreach(println)
    val data1 = sc.makeRDD(List(('a', 90), ('b', 78), ('c', 66), ('b', 78)), 2)
    val data2 = sc.makeRDD((List(('a', 5), ('b', 3), ('c', 4), ('a', 6), 2)))
    // data1.reduceByKey((x,y)=>x + y).foreach(println)
    //groupByKey根据key分组
    //data1.groupBy(_._1).foreach(println)
    // data1.groupBy().foreach(println)
    // data1.join(data2).foreach(println)
    //以左边为主，右边可能匹配不到
    // data1.leftOuterJoin(data2).foreach(println)
    //combineBykey
    data1.combineByKey(
      v => (v, 1),
      (t: (Int, Int), v) => {
        (t._1 + v, t._2 + 1)
      },
      (t1: (Int, Int), t2: (Int, Int)) => {
        (t1._1 + t2._1, t1._2 + t2._2)
      }
    ).foreach(println)
    //结束后释放资源
    sc.stop()

  }


}
